package cn.doitedu.rpc

import java.util.UUID

import akka.actor.{Actor, ActorSelection, ActorSystem, Props}
import com.typesafe.config.ConfigFactory
//导入时间单位
import scala.concurrent.duration._

//Worker也是一个Actor，也可以收发消息
class Worker(val masterHost: String, val masterPort: Int, var memory: Int, var cores: Int) extends Actor {

  var masterRef: ActorSelection = _

  //生成Worker的唯一ID
  val WORKER_ID = UUID.randomUUID().toString

  //所有的Actor，有生命周期方法（按照一定顺序，必须调用的方法）
  //在构造方法之后，在receive方法执行之前，一定会调用一次preStart方法
  override def preStart(): Unit = {
    //跟Master建立连接
    //选择与哪个ActorSystem下的Actor建立连接
    //akka.tcp://${ActorSystem名称}@${主机名}:@{端口号}/user/${Actor名称}
    //返回了与Master建立的连接（代理对象）
    masterRef = context.actorSelection(s"akka.tcp://${Master.MASTER_ACTOR_SYSTEM_NAME}@${masterHost}:${masterPort}/user/${Master.MASTER_ACTOR_NAME}")

    //Worker向Master发送注册消息
    //masterRef ! "register"
    masterRef ! RegisterWorker(WORKER_ID, memory, cores)
  }

  //接收消息，Receive是一个偏函数类型
  override def receive: Receive = {

    //Master返回给Worker的注册成功消息
    case RegisteredWorker => {
      //1.启动定时器，定期发送心跳
      //schedule定期调用指定的方法
      //Worker应该把心跳信息发送给Master，但是目前没法直接发送
      //导入隐式转换
      import context.dispatcher
      //self返回的是当前自己的Actor引用
      //定时器启动后，会每15秒钟，自己给自己发送一个SendHeartbeat消息
      context.system.scheduler.schedule(0 millisecond, 15000 millisecond, self, SendHeartbeat)

    }

    //Worker向Master定期发送的心跳消息
    case SendHeartbeat => {
      //在发送心跳之前，可以实现一些逻辑判断（省略...）
      //向Master发送心跳消息
      masterRef ! Heartbeat(WORKER_ID)
    }
  }
}

object Worker {

  val WORKER_ACTOR_SYSTEM_NAME = "WorkerActorSystem"
  val WORKER_ACTOR_NAME = "WorkerActor"

  def main(args: Array[String]): Unit = {


    val masterHost = args(0)
    val masterPort = args(1).toInt

    val workerHost = args(2)
    val workerPort = args(3).toInt

    val workerMemory = args(4).toInt
    val workerCores = args(5).toInt

    val configStr =
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname = $workerHost
         |akka.remote.netty.tcp.port = $workerPort
         |""".stripMargin
    val config = ConfigFactory.parseString(configStr)
    //创建ActorSystem
    val actorSystem = ActorSystem(WORKER_ACTOR_SYSTEM_NAME, config)
    //创建Actor
    actorSystem.actorOf(Props(new Worker(masterHost, masterPort, workerMemory, workerCores)), WORKER_ACTOR_NAME)

  }

}
